package com.silverbox.core.util;

import java.io.IOException;
import java.nio.charset.Charset;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;

import org.apache.commons.codec.binary.Base64;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.silverbox.sweeporder.core.thread.MqttThread;

public class AlibabaMQTT {
	/**
     * 设置当前用户私有的MQTT的接入点。例如此处示意使用XXX，实际使用请替换用户自己的接入点。接入点的获取方法是，在控制台申请MQTT实例，每个实例都会分配一个接入点域名。
     */
    final String broker ="tcp://mqtt-cn-45903fb6901.mqtt.aliyuncs.com:1883";
    /**
     * 设置阿里云的AccessKey，用于鉴权
     */
    final String acessKey ="m3Rvos1pt1Ziyp0p";
    /**
     * 设置阿里云的SecretKey，用于鉴权
     */
    final String secretKey ="7IO0iyhPh77eBCk9k14GBuJfjEBcbU";
    /**
     * 发消息使用的一级Topic，需要先在MQ控制台里申请
     */
    final String topic ="msg_yhbc";
    protected Logger log = (Logger) LoggerFactory.getLogger(getClass());
    String Gropid="GID-YHBC";
    MemoryPersistence persistence = new MemoryPersistence();
    final String clientId =Gropid+"@@@"+UUID.randomUUID().toString().replaceAll("-", "");
    MqttClient sampleClient ;
    boolean iscomplete=false;
    boolean success=true;
     /**
      * 
      * @param secondTopic
      * @return Boolean
      * @throws IOException
      * 
      */
    public  Boolean sendMqtt(String secondTopic,String content) throws IOException {
    	/**
         * MQTT的ClientID，一般由两部分组成，GroupID@@@DeviceID
         * 其中GroupID在MQ控制台里申请
         * DeviceID由应用方设置，可能是设备编号等，需要唯一，否则服务端拒绝重复的ClientID连接
         */
        //final String clientId =Gropid+"@@@"+clientid;
        String sign;
        //MemoryPersistence persistence = new MemoryPersistence();
        try {
        	sampleClient  = new MqttClient(broker, clientId, persistence);
            //sampleClient = new MqttClient(broker, clientId, persistence);
            final MqttConnectOptions connOpts = new MqttConnectOptions();
            System.out.println("Connecting to broker: " + broker);
            /**
             * 计算签名，将签名作为MQTT的password。
             * 签名的计算方法，参考工具类MacSignature，第一个参数是ClientID的前半部分，即GroupID
             * 第二个参数阿里云的SecretKey
             */
            sign = macSignature(clientId.split("@@@")[0], secretKey);
            connOpts.setUserName(acessKey);
            connOpts.setServerURIs(new String[] { broker });
            connOpts.setPassword(sign.toCharArray());
            connOpts.setCleanSession(false);
            connOpts.setKeepAliveInterval(90);
            sampleClient.setCallback(new MqttCallback() {
                public void connectionLost(Throwable throwable) {
                    //System.out.println("mqtt connection lost");
                    while(!sampleClient.isConnected()){
                        try {
                            Thread.sleep(1000);
                            sampleClient.connect(connOpts);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
                public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                    System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
                }
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                    System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId()+" isComplete="+iMqttDeliveryToken.isComplete());
                    iscomplete=true;
                    //启动线程关闭资源
                    MqttThread mth=new MqttThread(sampleClient);
                    Thread baseThread = new Thread(mth);
                    baseThread.start();
                }
            });
            sampleClient.connect(connOpts);
                try {
                    final MqttMessage message = new MqttMessage(content.getBytes("UTF-8"));
                    //System.out.println("bytes====="+content.getBytes("UTF-8")+"    "+content.getBytes().length);
                    message.setQos(1);
                    //System.out.println(" pushed at "+new Date()+" "+ content);
                    /**
                     *消息发送到某个主题Topic，所有订阅这个Topic的设备都能收到这个消息。
                     * 遵循MQTT的发布订阅规范，Topic也可以是多级Topic。此处设置了发送到二级Topic
                     */
                    sampleClient.publish(topic+"/"+secondTopic+"/", message);
                    /**
                     * 如果发送P2P消息，二级Topic必须是“p2p”，三级Topic是目标的ClientID
                     * 此处设置的三级Topic需要是接收方的ClientID
                     */
                    //String p2pTopic =topic+"/p2p/GID_mqttdelay3@@@DEVICEID_001";
                   //sampleClient.publish(p2pTopic,message);
                } catch (Exception e) {
                    e.printStackTrace();
                    success=false;
                    log.error("发送失败"+e.getMessage());
                }            
            
        } catch (Exception me) {
            me.printStackTrace();
            success=false;
            log.error("发送失败"+me.getMessage());
        }
        return success;
    }
    /**
     * 
     * @param secondTopic 二级topic
     * @param message 消息内容
     * @return
     * @throws Exception
     */
    public static Boolean send(String secondTopic,String message ) throws Exception {
		return new AlibabaMQTT().sendMqtt( secondTopic,message);
	}
    
    
    
    
    /**
     * 
     * @param secondTopic
     * @return Boolean
     * @throws IOException
     * 
     */
   public  Boolean sendObjctByMqtt(String secondTopic,Object content) throws IOException {
   	/**
        * MQTT的ClientID，一般由两部分组成，GroupID@@@DeviceID
        * 其中GroupID在MQ控制台里申请
        * DeviceID由应用方设置，可能是设备编号等，需要唯一，否则服务端拒绝重复的ClientID连接
        */
       //final String clientId =Gropid+"@@@"+clientid;
       String sign;
       //MemoryPersistence persistence = new MemoryPersistence();
       try {
       	sampleClient  = new MqttClient(broker, clientId, persistence);
           //sampleClient = new MqttClient(broker, clientId, persistence);
           final MqttConnectOptions connOpts = new MqttConnectOptions();
           System.out.println("Connecting to broker: " + broker);
           /**
            * 计算签名，将签名作为MQTT的password。
            * 签名的计算方法，参考工具类MacSignature，第一个参数是ClientID的前半部分，即GroupID
            * 第二个参数阿里云的SecretKey
            */
           sign = macSignature(clientId.split("@@@")[0], secretKey);
           connOpts.setUserName(acessKey);
           connOpts.setServerURIs(new String[] { broker });
           connOpts.setPassword(sign.toCharArray());
           connOpts.setCleanSession(false);
           connOpts.setKeepAliveInterval(90);
           sampleClient.setCallback(new MqttCallback() {
               public void connectionLost(Throwable throwable) {
                   //System.out.println("mqtt connection lost");
                   while(!sampleClient.isConnected()){
                       try {
                           Thread.sleep(1000);
                           sampleClient.connect(connOpts);
                       } catch (Exception e) {
                           e.printStackTrace();
                       }
                   }
               }
               public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                   System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
               }
               public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                   //System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId()+" isComplete="+iMqttDeliveryToken.isComplete());
                   iscomplete=true;
                   //启动线程关闭资源
                   MqttThread mth=new MqttThread(sampleClient);
                   Thread baseThread = new Thread(mth);
                   baseThread.start();
               }
           });
           sampleClient.connect(connOpts);
               try {
            	   
                   final MqttMessage message = new MqttMessage(BytesObjectUtils.toByteArray(content));
                   //System.out.println("bytes====="+content.getBytes("UTF-8")+"    "+content.getBytes().length);
                   message.setQos(1);
                   //System.out.println(" pushed at "+new Date()+" "+ content);
                   /**
                    *消息发送到某个主题Topic，所有订阅这个Topic的设备都能收到这个消息。
                    * 遵循MQTT的发布订阅规范，Topic也可以是多级Topic。此处设置了发送到二级Topic
                    */
                   sampleClient.publish(topic+"/"+secondTopic+"/", message);
                   /**
                    * 如果发送P2P消息，二级Topic必须是“p2p”，三级Topic是目标的ClientID
                    * 此处设置的三级Topic需要是接收方的ClientID
                    */
                   //String p2pTopic =topic+"/p2p/GID_mqttdelay3@@@DEVICEID_001";
                  //sampleClient.publish(p2pTopic,message);
               } catch (Exception e) {
                   e.printStackTrace();
                   success=false;
                   log.error("发送失败"+e.getMessage());
               }            
           
       } catch (Exception me) {
           me.printStackTrace();
           success=false;
           log.error("发送失败"+me.getMessage());
       }
       return success;
   }
    
    
    
    
    
    
    
    
    public static void main(String[] args) throws Exception {
    	String msg = "";
    	for(int i=0;i<5000;i++) {
    		msg = "{\"msg\":\"桌号：A001，顾客已付款+"+i+"\",\"orderType\":0,\"tableUid\":\"6D514C63DCAB9529FC4DB3533BF74B6E\",\"totalFee\":4.0,\"orderNum\":\"201902261155300086085817841\",\"type\":\"pay\",\"outOrderNum\":\"\",\"tableNo\":\"A001\"}";
    		new AlibabaMQTT().sendMqtt( "688756_1006823", msg);
    	}
    	
    }
    
    
    /************************MacSignature**********************************************/
    /**
     * @param text      要签名的文本
     * @param secretKey 阿里云MQ SecretKey
     * @return 加密后的字符串
     * @throws InvalidKeyException
     * @throws NoSuchAlgorithmException
     */
    public static String macSignature(String text, String secretKey) throws InvalidKeyException, NoSuchAlgorithmException {
        Charset charset = Charset.forName("UTF-8");
        String algorithm = "HmacSHA1";
        Mac mac = Mac.getInstance(algorithm);
        mac.init(new SecretKeySpec(secretKey.getBytes(charset), algorithm));
        byte[] bytes = mac.doFinal(text.getBytes(charset));
        return new String(Base64.encodeBase64(bytes), charset);
    }
    /**
     * 发送方签名方法
     *
     * @param clientId  MQTT ClientID
     * @param secretKey 阿里云MQ SecretKey
     * @return 加密后的字符串
     * @throws NoSuchAlgorithmException
     * @throws InvalidKeyException
     */
    public static String publishSignature(String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
        return macSignature(clientId, secretKey);
    }
    /**
     * 订阅方签名方法
     *
     * @param topics    要订阅的Topic集合
     * @param clientId  MQTT ClientID
     * @param secretKey 阿里云MQ SecretKey
     * @return 加密后的字符串
     * @throws NoSuchAlgorithmException
     * @throws InvalidKeyException
     */
    public static String subSignature(List<String> topics, String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
        Collections.sort(topics); //以字典顺序排序
        String topicText = "";
        for (String topic : topics) {
            topicText += topic + "\n";
        }
        String text = topicText + clientId;
        return macSignature(text, secretKey);
    }
    /**
     * 订阅方签名方法
     *
     * @param topic     要订阅的Topic
     * @param clientId  MQTT ClientID
     * @param secretKey 阿里云MQ SecretKey
     * @return 加密后的字符串
     * @throws NoSuchAlgorithmException
     * @throws InvalidKeyException
     */
    public static String subSignature(String topic, String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
        List<String> topics = new ArrayList<String>();
        topics.add(topic);
        return subSignature(topics, clientId, secretKey);
    }
}
